// -*-mode:c++; coding:utf-8-*-

#include <sstream>

#include <logger.hpp>

#include "sstable_block_cache_manager.hpp"

namespace xbase 
{

	namespace details
	{
		sstable_block_cache_manager_impl_fast *sstable_block_cache_manager_impl_fast::s_instance = NULL;

		sstable_block_cache_manager_impl_fast::~sstable_block_cache_manager_impl_fast()
		{
			boost::mutex::scoped_lock lock(m_mutex);
			typedef ref_count_index_type::iterator it_type;
			const it_type end = m_block_cache.get<ref_count>().end();
			for(it_type it = m_block_cache.get<ref_count>().begin();
			    it != end; ++it)
			{
				if(it->ref_count() != 0)
				{
					LOG(FATAL, "sstable_block_cache_manager_impl_fast::~sstable_block_cache_manager_impl_fast failed on block "
					    << it->m_node->to_string() << ", dump: \n" << to_string());
					assert(false);
				}
				deallocate(it->m_node);
			}
		}

		bool sstable_block_cache_manager_impl_fast::commit_with_ref_count(sstable_block_cache_manager_impl_fast::cache_node_type *node,
										  std::size_t ref_count_value)
		{
			const uint64_t request_size = node->m_value.size();
			if(request_size > m_max_memory)
			{
				LOG(ERROR, "sstable_block_cache_manager_impl_fast::"
				    "commit_with_ref_count node size "
				    << node->m_value.size() << " too big against max memory "
				    << m_max_memory << ", node = " << node->to_string());
				return false;
			}

			boost::mutex::scoped_lock lock(m_mutex);
			key_index_type &key_index = m_block_cache.get<key>();

			const key_index_type::iterator iter = key_index.find(node->m_key);
			if(iter != key_index.end())
			{
				LOG(ERROR, "sstable_block_cache_manager_impl_fast::commit_with_ref_count "
				    "node already exists, commit node = "
				    << node->to_string() << ", cached node = " << iter->m_node->to_string());
				return false;
			}

			ref_count_index_type &ref_count_index = m_block_cache.get<ref_count>();
			if(m_free_memory < request_size) // 未用内存不足，则看可用内存是否足够
			{
				if(available_memory() < request_size) // 可用内存也不够，则返回失败
				{
					LOG(WARN, "sstable_block_cache_manager_impl_fast::commit_with_ref_count"
					    " cache memory too small, request size = "
					    << request_size << ", cur free memory = " << m_free_memory
					    << ", busy memory = " << m_busy_memory
					    << ", cache memory size = " << m_max_memory);
					return false;
				}
				// 淘汰后可以放得下，开始淘汰...
				const ref_count_index_type::iterator begin = ref_count_index.begin();
				const ref_count_index_type::iterator end = ref_count_index.end();
				ref_count_index_type::iterator pos = begin;
				for(;(m_free_memory < request_size) && (pos != end) && (pos->ref_count() == 0);
				    ++pos)
				{
					// 这只是清除未使用的块，不影响busy_memory的大小
					m_free_memory += pos->m_node->m_value.size();
					--m_block_number;

					LOG(TRACE, "sstable_block_cache_manager_impl_fast::commit_with_ref_count "
					    "removing cache node " << pos->m_node->to_string());
					deallocate(pos->m_node);
				}
				ref_count_index.erase(begin, pos);

				if(m_free_memory < request_size) // 若仍不足，则是成员变量更新错误，是有BUG！
				{
					LOG(FATAL, "sstable_block_cache_manager_impl_fast::commit_with_ref_count"
					    " BAD member values");
					assert(false);
					return false;
				}
			}

			cache_node_wrapper wr(node);
			wr.reset_ref_count(ref_count_value);
			const std::pair<ref_count_index_type::iterator, bool> ret = ref_count_index.insert(wr);
			if(ret.second == false)
			{
				LOG(ERROR, "sstable_block_cache_manager_impl_fast::commit_with_ref_count"
				    " insert into block cache failed, collide with "
				    << ret.first->m_node->to_string());
				return false;
			}

			m_free_memory -= request_size;
			if(ref_count_value != 0)
			{
				m_busy_memory += request_size;
				m_max_busy_memory = (std::max)(m_max_busy_memory, m_busy_memory);
			}
			m_block_number += 1;
			LOG(TRACE, "sstable_block_cache_manager_impl_fast::commit_with_ref_count "
			    "commit ok, node = " << node->to_string());
			return true;
		}

		sstable_block_cache_manager_impl_fast::cache_node_type *
		sstable_block_cache_manager_impl_fast::checkout(const sstable_block_cache_manager_impl_fast::cache_key_type &cache_key)
		{
			boost::mutex::scoped_lock lock(m_mutex);
			++ m_checkout_counter;
			key_index_type &key_index = m_block_cache.get<key>();
			const key_index_type::iterator iter = key_index.find(cache_key);
			if(iter == key_index.end())
			{
				return NULL;
			}
			cache_node_wrapper wr = *iter;
			wr.inc_ref_count();
			const bool ok = key_index.replace(iter, wr);
			if(ok == false)
			{
				LOG(ERROR, "sstable_block_cache_manager_impl_fast::checkout replace cache node failed, node = "
				    << iter->m_node->to_string());
				return NULL;
			}
			if(wr.ref_count() == 1) // 刚从未使用转为被使用状态
			{
				m_busy_memory += wr.m_node->m_value.size();
				m_max_busy_memory = (std::max)(m_max_busy_memory, m_busy_memory);
			}
			++ m_checkout_counter_ok;
			return wr.m_node;
		}

		bool sstable_block_cache_manager_impl_fast::checkin(cache_node *node)
		{
			boost::mutex::scoped_lock lock(m_mutex);
			key_index_type &key_index = m_block_cache.get<key>();
			const key_index_type::iterator iter = key_index.find(node->m_key);
			if(iter == key_index.end())
			{
				return false;
			}
			cache_node_wrapper wr = *iter;
			wr.dec_ref_count();
			const bool ok = key_index.replace(iter, wr);
			if(ok == false)
			{
				LOG(ERROR, "sstable_block_cache_manager_impl_fast::checkin replace cache node failed, node = "
				    << iter->m_node->to_string());
				return false;
			}
			if(wr.ref_count() == 0) // 刚从被使用转为未使用状态
			{
				m_busy_memory -= wr.m_node->m_value.size();
				// 可以确定busy_memory变小了，所以不用更新该值
				//m_max_used_memory = (std::max)(m_max_used_memory, m_busy_memory);
			}
			return true;
		}

#ifndef NDEBUG
		void sstable_block_cache_manager_impl_fast::clear_sstable_cache(file_id_type id)
		{
			boost::mutex::scoped_lock lock(m_mutex);
			file_id_index_type &file_id_index = m_block_cache.get<file_id>();
			const std::pair<file_id_index_type::iterator, file_id_index_type::iterator>
				range = file_id_index.equal_range(id);
			for(file_id_index_type::iterator it = range.first;
			    it != range.second; ++it)
			{
				if(it->ref_count() != 0)
				{
					LOG(FATAL, "sstable_block_cache_manager_impl_fast::clear_sstable_cache check failed on block "
					    << it->m_node->to_string() << ", file id = " << id
					    << ", sstable_block_cache_manager_impl_fast dump: \n"
					    << to_string());
					assert(false);
				}
				LOG(TRACE, "sstable_block_cache_manager_impl_fast::clear_sstable_cache "
				    "removing cache node " << it->m_node->to_string());
				// 这只是清除未使用的块，不影响busy_memory的大小
				m_free_memory += it->m_node->m_value.size();
				--m_block_number;
				deallocate(it->m_node);
			}
			file_id_index.erase(range.first, range.second);
		}
#endif	// NDEBUG

		std::string sstable_block_cache_manager_impl_fast::to_string() const
		{
			std::ostringstream out;
			out << "{max_memory: " << m_max_memory
			    << "; free_memory: " << m_free_memory
			    << "; busy_memory: " << m_busy_memory
			    << "; max_busy_memory: " << m_max_busy_memory
			    << "; block_number: " << m_block_number
			    << "; checkout_counter: " << m_checkout_counter
			    << "; checkout_counter_ok: " << m_checkout_counter_ok
			    << "; hit_rate: " << ((double)(std::size_t)(hit_rate() * 100000.0)) / 1000.0 << "%"
			    << "; block_cache: {\n";
			const ref_count_index_type &index = m_block_cache.get<ref_count>();
			typedef ref_count_index_type::const_iterator it_type;
			std::size_t no = 0;
			for(it_type it = index.begin(); it != index.end(); ++it, ++no)
			{
				out << "\t[";
				out.width(5);
				out << no << "] ";
				out << it->m_node->to_string() << "\n";
			}
			out << "};}";
			return out.str();
		}

	} // namespace details

} // namespace xbase


